package com.mpgame.common.helper;

import java.util.List;

import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import com.mpgame.common.proto.CmdId;
import com.mpgame.common.proto.ErrorCode;
import com.mpgame.common.proto.StreamPacket;
import com.mpgame.common.utils.Log;

import io.grpc.stub.StreamObserver;

public class GrpcStreamHelper {
    private volatile StreamObserver<StreamPacket> obs;
    private final Object sendLock = new Object();

    public GrpcStreamHelper(StreamObserver<StreamPacket> obs) {
        this.obs = obs;
    }

    public StreamObserver<StreamPacket> getObs() {
        return this.obs;
    }

    public void setObs(StreamObserver<StreamPacket> obs) {
        this.obs = obs;
    }

    public void sendStreamPkt(int targetAccountId, CmdId msgType, Message msg) {
        synchronized (sendLock) {
            if (this.obs != null) {
                try {
                    StreamPacket pkt = StreamPacket.newBuilder()
                            .setCmd(msgType)
                            .setEc(ErrorCode.SUCCESS)
                            .addTargetAccountId(targetAccountId)
                            .setData(msg.toByteString())
                            .build();
                    obs.onNext(pkt);
                } catch (Exception e) {
                    Log.err("发送流消息失败：" + e.getMessage());
                    obs = null;
                }
            } else {
                Log.warn("流连接不存在，无法发送消息！");
            }
        }
    }

    public void sendStreamPkt(List<Integer> targetAccountIds, CmdId msgType, Message msg) {
        synchronized (sendLock) {
            if (this.obs != null) {
                try {
                    StreamPacket pkt = StreamPacket.newBuilder()
                            .setCmd(msgType)
                            .setEc(ErrorCode.SUCCESS)
                            .addAllTargetAccountId(targetAccountIds)
                            .setData(msg.toByteString())
                            .build();
                    obs.onNext(pkt);
                } catch (Exception e) {
                    Log.err("群发送流消息失败：" + e.getMessage());
                    obs = null;
                }
            } else {
                Log.warn("流连接不存在，无法发送群消息！");
            }
        }
    }

    public void sendErrStreamPkt(int targetAccountId, CmdId msgType, String errMsg) {
        synchronized (sendLock) {
            if (this.obs != null) {
                try {
                    StreamPacket pkt = StreamPacket.newBuilder()
                            .setCmd(msgType)
                            .setEc(ErrorCode.INVALID_PARAM)
                            .addTargetAccountId(targetAccountId)
                            .setData(ByteString.copyFromUtf8(errMsg))
                            .build();
                    obs.onNext(pkt);
                } catch (Exception e) {
                    Log.err("发送流错误消息失败：" + e.getMessage());
                    obs = null;
                }
            } else {
                Log.warn("流连接不存在，无法发送错误消息！");
            }
        }
    }

    public void sendErrStreamPkt(List<Integer> targetAccountIds, CmdId msgType, String errMsg) {
        synchronized (sendLock) {
            if (this.obs != null) {
                try {
                    StreamPacket pkt = StreamPacket.newBuilder()
                            .setCmd(msgType)
                            .setEc(ErrorCode.INVALID_PARAM)
                            .addAllTargetAccountId(targetAccountIds)
                            .setData(ByteString.copyFromUtf8(errMsg))
                            .build();
                    obs.onNext(pkt);
                } catch (Exception e) {
                    Log.err("发送流错误消息失败：" + e.getMessage());
                    obs = null;
                }
            } else {
                Log.warn("流连接不存在，无法发送错误消息！");
            }
        }
    }
}
